
/*-------------------------------------------------------------------------
 *
 * blscan.c
 *		Bloom index scan functions.
 *
 * Copyright (c) 2016-2022, PostgreSQL Global Development Group
 *
 * IDENTIFICATION
 *	  contrib/bloom/blscan.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include "access/relscan.h"
#include "bloom.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "storage/bufmgr.h"
#include "storage/lmgr.h"
#include "utils/memutils.h"
#include "utils/rel.h"

/*
 * 开始扫描布隆索引。
 */
IndexScanDesc blbeginscan(Relation fc_r, int fc_nkeys, int fc_norderbys)
{
	IndexScanDesc fc_scan;
	BloomScanOpaque fc_so;

	fc_scan = RelationGetIndexScan(fc_r, fc_nkeys, fc_norderbys);

	fc_so = (BloomScanOpaque) palloc(sizeof(BloomScanOpaqueData));
	initBloomState(&fc_so->state, fc_scan->indexRelation);
	fc_so->sign = NULL;

	fc_scan->opaque = fc_so;

	return fc_scan;
}

/*
 * 重新扫描布隆索引。
 */
void blrescan(IndexScanDesc fc_scan, ScanKey fc_scankey, int fc_nscankeys,
		 ScanKey fc_orderbys, int fc_norderbys)
{
	BloomScanOpaque fc_so = (BloomScanOpaque) fc_scan->opaque;

	if (fc_so->sign)
		pfree(fc_so->sign);
	fc_so->sign = NULL;

	if (fc_scankey && fc_scan->numberOfKeys > 0)
	{
		memmove(fc_scan->keyData, fc_scankey,
				fc_scan->numberOfKeys * sizeof(ScanKeyData));
	}
}

/*
 * 结束扫描布隆索引。
 */
void blendscan(IndexScanDesc fc_scan)
{
	BloomScanOpaque fc_so = (BloomScanOpaque) fc_scan->opaque;

	if (fc_so->sign)
		pfree(fc_so->sign);
	fc_so->sign = NULL;
}

/*
 * 将所有匹配的元组插入位图中。
 */
int64 blgetbitmap(IndexScanDesc fc_scan, TIDBitmap *fc_tbm)
{
	int64		fc_ntids = 0;
	BlockNumber fc_blkno = BLOOM_HEAD_BLKNO,
				fc_npages;
	int			fc_i;
	BufferAccessStrategy fc_bas;
	BloomScanOpaque fc_so = (BloomScanOpaque) fc_scan->opaque;

	if (fc_so->sign == NULL)
	{
		/* 新的搜索：必须计算搜索签名 */
		ScanKey		fc_skey = fc_scan->keyData;

		fc_so->sign = palloc0(sizeof(BloomSignatureWord) * fc_so->state.opts.bloomLength);

		for (fc_i = 0; fc_i < fc_scan->numberOfKeys; fc_i++)
		{
			/*
			 * 假设布隆索引可索引操作符是严格的，因此对于NULL键无法找到任何东西。
			 */
			if (fc_skey->sk_flags & SK_ISNULL)
			{
				pfree(fc_so->sign);
				fc_so->sign = NULL;
				return 0;
			}

			/* 将下一个值添加到签名中 */
			signValue(&fc_so->state, fc_so->sign, fc_skey->sk_argument,
					  fc_skey->sk_attno - 1);

			fc_skey++;
		}
	}

	/*
	 * 我们将读取整个索引。这就是我们使用适当的缓冲区访问策略的原因。
	 */
	fc_bas = GetAccessStrategy(BAS_BULKREAD);
	fc_npages = RelationGetNumberOfBlocks(fc_scan->indexRelation);
	pgstat_count_index_scan(fc_scan->indexRelation);

	for (fc_blkno = BLOOM_HEAD_BLKNO; fc_blkno < fc_npages; fc_blkno++)
	{
		Buffer		fc_buffer;
		Page		fc_page;

		fc_buffer = ReadBufferExtended(fc_scan->indexRelation, MAIN_FORKNUM,
									fc_blkno, RBM_NORMAL, fc_bas);

		LockBuffer(fc_buffer, BUFFER_LOCK_SHARE);
		fc_page = BufferGetPage(fc_buffer);
		TestForOldSnapshot(fc_scan->xs_snapshot, fc_scan->indexRelation, fc_page);

		if (!PageIsNew(fc_page) && !BloomPageIsDeleted(fc_page))
		{
			OffsetNumber fc_offset,
						fc_maxOffset = BloomPageGetMaxOffset(fc_page);

			for (fc_offset = 1; fc_offset <= fc_maxOffset; fc_offset++)
			{
				BloomTuple *fc_itup = BloomPageGetTuple(&fc_so->state, fc_page, fc_offset);
				bool		fc_res = true;

				/* 使用扫描签名检查索引签名 */
				for (fc_i = 0; fc_i < fc_so->state.opts.bloomLength; fc_i++)
				{
					if ((fc_itup->sign[fc_i] & fc_so->sign[fc_i]) != fc_so->sign[fc_i])
					{
						fc_res = false;
						break;
					}
				}

				/* 将匹配的元组添加到位图 */
				if (fc_res)
				{
					tbm_add_tuples(fc_tbm, &fc_itup->heapPtr, 1, true);
					fc_ntids++;
				}
			}
		}

		UnlockReleaseBuffer(fc_buffer);
		CHECK_FOR_INTERRUPTS();
	}
	FreeAccessStrategy(fc_bas);

	return fc_ntids;
}
